package com.google.code.p.restrpc.spring.heartbeat;

import java.util.Date;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.google.code.p.restrpc.core.RpcServiceRepository;
import com.google.code.p.restrpc.core.thread.NamedThreadFactory;
import com.google.code.p.restrpc.spring.RpcServiceRemoteExporter;

/**
 * @author dz
 */
public class ExportHeartBeater implements Runnable {
    
    private static Log logger = LogFactory.getLog(ExportHeartBeater.class);
    
    private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1,
                                                                                                      new NamedThreadFactory("restrpc-exporter-heartbeat",
                                                                                                                             true));
    
    private RpcServiceRemoteExporter exporter;
    
    // 心跳定时器
    private ScheduledFuture heatbeatTimer;
    
    // 心跳超时，毫秒。缺省0，不会执行心跳。
    private int heartbeat;
    
    private AtomicBoolean started = new AtomicBoolean(false);
    
    public ExportHeartBeater(RpcServiceRemoteExporter exporter,
                                  int heartbeat) {
        this.exporter = exporter;
        this.heartbeat = heartbeat;
        start();
    }
    
    /**
     * 
     */
    public void start() {
        if (started.get()) {
            logger.warn("The HeartBeatExportMonitor has been started!");
            return;
        }
        if (heartbeat <= 0) {
            logger.warn("heartbeat < 0, the HeartBeatExportMonitor would not start!");
            return;
        }
        
        heatbeatTimer = scheduledThreadPoolExecutor.scheduleWithFixedDelay(this,
                                                                           heartbeat,
                                                                           heartbeat,
                                                                           TimeUnit.MILLISECONDS);
        started.compareAndSet(false, true);
        if (logger.isDebugEnabled()) {
            logger.debug("restrpc-exporter-heartbeat is started!");
        }
    }
    
    /**
     * 
     */
    public void stop() {
        if (!started.get()) {
            logger.warn("The HeartBeatExportMonitor has been stoped!");
            return;
        }
        if (heatbeatTimer != null && !heatbeatTimer.isCancelled()) {
            try {
                heatbeatTimer.cancel(true);
                scheduledThreadPoolExecutor.purge();
            }
            catch (Throwable e) {
                logger.warn(e.getMessage(), e);
            }
        }
        heatbeatTimer = null;
        started.compareAndSet(true, false);
        if (logger.isDebugEnabled()) {
            logger.debug("restrpc-exporter-heartbeat is stoped!");
        }
    }
    
    /**
     * 
     */
    public void run() {
        if (logger.isTraceEnabled()) {
            logger.trace(new Date() + " begin...");
        }
        
        if (heartbeat <= 0) {
            return;
        }
        
        if (!RpcServiceRepository.getInstance().isExported()) {
            return;
        }
        
        try {
            exporter.getRegistryClient().activate(exporter.getAddress(),
                                                  exporter.getPort());
        }
        catch (Exception e) {
            logger.error(e, e);
        }
        
        if (logger.isTraceEnabled()) {
            logger.trace(new Date() + " end...");
        }
    }
    
    @Override
    protected void finalize() throws Throwable {
        stop();
        try {
            scheduledThreadPoolExecutor.shutdown();
        }
        catch (Exception e) {
        }
    }
    
}
